package consumer

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"

	"github.com/dimchansky/utfbom"
	jsoniter "github.com/json-iterator/go"
	"github.com/jszwec/csvutil"
	"github.com/nsqio/go-nsq"
	"github.com/xuri/excelize/v2"
	"zhonghui/console/structs"
	"zhonghui/console/warehouse/cfgx"
	"zhonghui/console/warehouse/dbx"
	"zhonghui/console/warehouse/dbx/db_zh_kernel"
	"zhonghui/console/warehouse/helperx"
	"zhonghui/console/warehouse/logx"
	"zhonghui/console/warehouse/modelx/model_zh_kernel"
	"zhonghui/console/warehouse/servicex"
)

// FileParseConsumer 处理文件解析
type FileParseConsumer struct{}

type fileData struct {
	ID     int    `csv:"id"`
	Phone  string `csv:"phone"`
	Age    int    `csv:"age"`
	Gender int    `csv:"gender"`
	Name   string `csv:"name"`
}

func (h *FileParseConsumer) GetTopicName() string {
	return cfgx.Cfg().GetString("console.topics.file_parse")
}

func (h *FileParseConsumer) GetQueueName() string {
	return cfgx.Cfg().GetString("console.channels.file_parse")
}

func (h *FileParseConsumer) HandleMessage(m *nsq.Message) error {
	defer func() {
		if r := recover(); r != nil {
			logx.Zap().Errorw("消费者异常", "err", errors.New(fmt.Sprintf("%s", r)))
		}
	}()

	ctx := helperx.MakeCtxWithRequestId()

	msgId := fmt.Sprintf("%s", m.ID)

	logx.ZapTrace(ctx).Infow("开始处理消息", "queue_name", h.GetQueueName(), "msg_id", msgId, "msg_body", string(m.Body))

	body := structs.FileParseWorkerBody{}
	if err := jsoniter.Unmarshal(m.Body, &body); err != nil {
		logx.ZapTrace(ctx).Errorw("解析通知结构失败，忽略", "queue_name", h.GetQueueName(), "msg_id", msgId, "err", err)
		return nil
	}

	// 查询任务
	task := model_zh_kernel.SysTask{}
	if err := db_zh_kernel.ClientCtx(ctx).Where("id = ? and status = ?", body.TaskID, structs.TaskStatusCreated).Take(&task).Error; err != nil {
		return nil
	}

	data, err := h.parseTable(ctx, task.Source)
	if err != nil {
		logx.ZapTrace(ctx).Errorw("解析文件失败，忽略", "queue_name", h.GetQueueName(), "msg_id", msgId, "err", err)
		h.fail(ctx, &task, "解析文件失败")
		return nil
	}

	// 保存手机号记录
	batchInc := make([]model_zh_kernel.SysSmsRecord, 0)
	dup := make(map[string]int)
	for _, d := range data {
		if _, ok := dup[d.Phone]; ok {
			continue
		}

		j := h.genParam(&task, &d)

		newRecord := model_zh_kernel.SysSmsRecord{
			TaskID: task.ID,
			Phone:  d.Phone,
			Status: "",
			Params: j,
		}
		batchInc = append(batchInc, newRecord)
		dup[d.Phone] = 1
	}
	db_zh_kernel.ClientCtx(ctx).CreateInBatches(batchInc, 1000)

	// 开始保存任务数据
	conn := dbx.NewTx(db_zh_kernel.ClientCtx(ctx))
	defer conn.End()

	var total int64
	err = conn.Tx.Model(model_zh_kernel.SysSmsRecord{}).Where("task_id = ?", task.ID).Count(&total).Error
	if err != nil {
		return err
	}

	err = conn.Tx.Where("id = ?", task.ID).Updates(model_zh_kernel.SysTask{
		Status: structs.TaskStatusPrepared,
		Count:  uint64(total),
	}).Error
	if err != nil {
		h.fail(ctx, &task, "解析文件失败")
		return err
	}

	conn.Commit()

	return nil
}

func (h *FileParseConsumer) parseTable(ctx context.Context, ossObject string) ([]fileData, error) {
	// 下载
	oc, err := servicex.OssClient("normal")
	if err != nil {
		logx.ZapTrace(ctx).Errorw("获取oss客户端失败", "queue_name", h.GetQueueName(), "err", err)
		return nil, err
	}
	s, err := oc.OssDownloadStream(ossObject)
	if err != nil {
		logx.ZapTrace(ctx).Errorw("解析文件失败", "queue_name", h.GetQueueName(), "err", err)
		return nil, err
	}

	// 解析
	if strings.HasSuffix(ossObject, ".csv") {
		return h.parseTableInCsv(ctx, s)
	} else if strings.HasSuffix(ossObject, ".xlsx") {
		return h.parseTableInXlsx(ctx, s)
	}

	return nil, errors.New("invalid file type")
}

func (h *FileParseConsumer) parseTableInCsv(ctx context.Context, s io.Reader) ([]fileData, error) {
	data := make([]fileData, 0)

	// 读取+去掉BOM
	content, _ := io.ReadAll(utfbom.SkipOnly(s))

	if err := csvutil.Unmarshal(content, &data); err != nil {
		logx.ZapTrace(ctx).Errorw("解析csv表格出错", "err", err)
		return nil, err
	}

	return data, nil
}

func (h *FileParseConsumer) parseTableInXlsx(ctx context.Context, s io.Reader) ([]fileData, error) {
	data := make([]fileData, 0)

	f, err := excelize.OpenReader(s)
	defer f.Close()

	rows, err := f.GetRows("Sheet1")
	if err != nil {
		logx.ZapTrace(ctx).Errorw("解析表格出错", "err", err)
		return nil, err
	}
	for i, row := range rows {
		if i < 2 {
			continue
		}
		tm := fileData{}
		for ii, colCell := range row {
			switch ii {
			case 0:
				tm.ID, _ = strconv.Atoi(colCell)
			case 1:
				tm.Phone = colCell
			case 2:
				tm.Age, _ = strconv.Atoi(colCell)
			case 3:
				tm.Name = colCell
			case 4:
				tm.Gender, _ = strconv.Atoi(colCell)
			}

			// 检查数据
			if tm.ID <= 0 {
				continue
			}
			if !helperx.IsPhone(tm.Phone) {
				continue
			}
		}
		data = append(data, tm)
	}

	return data, nil
}

func (h *FileParseConsumer) fail(ctx context.Context, task *model_zh_kernel.SysTask, extra string) {
	db_zh_kernel.ClientCtx(ctx).Model(model_zh_kernel.SysTask{}).Where("id = ? and status = 'created'", task.ID).Updates(map[string]interface{}{
		"status": "stopped",
		"remark": extra,
	})
}

func (h *FileParseConsumer) genParam(task *model_zh_kernel.SysTask, d *fileData) string {
	param := make([]string, 0)

	switch task.SmsTemplate {
	case "cfe5c2cc7e734a90b7cb189a8641a9cb":
		// 特殊模板固定参数
		param = append(param, "“惠孕保”")
		param = append(param, "bmegca")
	default:
		name := "您好"
		if d.Name != "" {
			lastName := string([]rune(d.Name)[:1])
			if d.Gender == 1 {
				name = lastName + "先生"
			} else if d.Gender == 2 {
				name = lastName + "女士"
			} else {
				name = strings.TrimSpace(d.Name)
			}
		}
		// 称呼
		param = append(param, name)
	}

	j, _ := jsoniter.MarshalToString(param)

	return j
}
